package com.youyu.core.jms.kafka;

import com.alibaba.fastjson.JSONObject;
import com.youyu.core.jms.kafka.model.ConsumerData;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
import org.yx.conf.AppInfo;
import org.yx.exception.BizException;
import org.yx.log.Log;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * <p>Author:  dxj</p>
 * <p>Date:2017/8/31 11:18</p>
 * <p>Description:抽象consumer高级Api</p>
 * <p>Copyright: Copyright (c) 2017</p>
 * <p>Company: youyu</p>
 */
public abstract class AbstractHighConsumer extends Thread implements IConsumer{

    @Override
    public String getConsumerName(){
        return setGroupId();
    }

    /**
     * 初始化
     */
    public ConsumerConnector createConsumer(String topic,String groupId,String customerId){
        Properties properties = new Properties();
        properties.put("topic",topic);
        properties.put("zookeeper.connect", AppInfo.get("sumk.zkurl"));
        properties.put("group.id", groupId);
        properties.put("consumer.id",customerId);
        properties.put("zookeeper.session.timeout.ms", AppInfo.get("kafka.zookeeper.session.timeout.ms","4000"));
        properties.put("zookeeper.sync.time.ms", AppInfo.get("kafka.zookeeper.sync.time.ms","200"));
        properties.put("auto.commit.interval.ms",AppInfo.get("kafka.auto.commit.interval.ms","1000"));
        properties.put("auto.offset.reset",AppInfo.get("kafka.auto.offset.reset","smallest"));
        properties.put("serializer.class",AppInfo.get("kafka.serializer.class","kafka.serializer.StringEncoder"));
        ConsumerConfig consumerConfig = new ConsumerConfig(properties);
        return kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);
    }

    @Override
    public void run(){
        String topic = setTopic();
        String groupId = setGroupId();
        String consumerId = setConsumerId();
        //初始化
        ConsumerConnector consumer = createConsumer(topic, groupId,consumerId);
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1));

        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

        Map<String, List<KafkaStream<String, String>>> consumerMap =
                consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
        //获得topic的数据流
        KafkaStream<String, String> stream = consumerMap.get(topic).get(0);
        ConsumerIterator<String, String> it = stream.iterator();
        // 遍历KafkaStream hasNext会block
        while (it.hasNext()) {
            // 获得MessageAndMetadata对象
            MessageAndMetadata<String, String> mam = it.next();
            try {
                if(!mam.message().isEmpty()){
                    process(mam.message());
                }else{
                    Log.get(AbstractHighConsumer.class).info("队列["+topic+"]发生空消息！");
                }
            } catch (Exception e) {
                Log.get(AbstractHighConsumer.class).info("队列["+topic+"]消费消息--"+mam.message()+"--失败！"+e.getMessage());
            }
        }
    }

    /**
     * 设置主题
     * @return
     * @throws org.yx.exception.BizException
     */
    public abstract String setTopic() throws BizException;

    /**
     * 设置消息群组
     * @return
     * @throws org.yx.exception.BizException
     */
    public abstract String setGroupId() throws BizException;

    /**
     * 设置消费者id
     * @return
     * @throws BizException
     */
    public abstract String setConsumerId() throws BizException;

    /**
     * 处理核心业务逻辑
     * @param message
     * @throws BizException
     */
    public abstract void process(String message) throws BizException;
}
